package com.wise.mqConsumer;

import com.rabbitmq.client.Channel;
import com.wise.mapper.UserMapper;
import com.wise.pojo.MsgConstants;
import com.wise.pojo.Person;
import com.wise.pojo.User;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @author rzc
 * @date 2021/9/22
 */
@Component
@Slf4j
public class PersonComsumer {
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private UserMapper userMapper;

    @RabbitListener(queues = MsgConstants.MAIL_QUEUE_NAME)
    @RabbitHandler
    public void handle(Message message, Channel channel){
        Person person = (Person) message.getPayload();

        MessageHeaders headers = message.getHeaders();
        //消息序列号
        long tag =(long)headers.get(AmqpHeaders.DELIVERY_TAG);
        // 获取消息msgId
        String msgId = (String) headers.get("spring_returned_message_correlation");
        HashOperations hashOperations = redisTemplate.opsForHash();
        try {
            if (hashOperations.entries("msg_log").containsKey(msgId)){
                log.error("==============>消息已经被消费",msgId);
                channel.basicAck(tag,false);
                return;
            }
            //根据人员创建用户
            User user = new User();
            user.setCode(person.getCode());
            user.setName(person.getName());
            userMapper.insert(user);
            log.info("=====================>用户创建成功",msgId);
            //将消息msgId存入redis
            hashOperations.put("msg_log",msgId,"OK");
            //手动确认消息
            channel.basicAck(tag,false);
        }catch (Exception e){
            try {
                /**
                 * 手动确认消息
                 * tag：消息序号
                 * multiple： 是否确认多条
                 * requeue： 是否回退到队列
                 */
                channel.basicNack(tag,false,true);
            } catch (IOException ex) {
                ex.printStackTrace();
            }
            log.error("=====================>用户创建失败失败",e.getMessage());
        }

    }
}
